//
// Created by Suzhen on 2017/2/24.
//

#include "FileStream.h"

USING_ORION_V2

FileStream::~FileStream() {
    SYS_FAULT(openInHandle_ == 0) << "stream依然被底层对象持有,发生内存泄露!";
    if (!dataWillSend_.empty()) {
#ifdef ENABLE_LOG
        LOG_I << "data not empty, 仍然有数据没有写出";
#endif
    }
}

FileStream::FileStream(FdType fd,
                       const IOBehavior &ioBehavior,
                       uint readTimeOut)
        : AbstractAsyncStream(ioBehavior),
          idleTimeMs_(readTimeOut) {
    coreAsyncPollData.fd = fd;
}

bool FileStream::isClosed() {
    return isClosed_;
}

bool FileStream::isClosing() {
    return isClosing_;
}

Future<bool> FileStream::sendAsync(const RawDataType *buf, size_t bufSz, bool transferData) {
    return sendAsync(IOBuffer(buf, bufSz, transferData));
}

Future<bool> FileStream::sendAsync(IOBuffer &&dataBuffer) {
    if (dataBuffer.len == 0) {
        return makeReadyThreadFuture(NO);
    }
    std::lock_guard<std::recursive_mutex> guard(dataLock_);
    Future<bool> completeFuture = dataBuffer.getCompleteFuture();
    if (isClosing_ ||
        isClosed_ ||
        uv_is_closing((uv_handle_t *) &coreAsyncPollData.pollCtx)) {
        return completeFuture;
    }
    dataWillSend_.push_back(std::move(dataBuffer));
    if (getCurrentThreadId() == coreAsyncPollData.tid) {
        FileStream::AsyncSendCallBack(&sendAsyncCtx_);
    } else {
        uv_async_send(&sendAsyncCtx_);
    }
    return completeFuture;
}

Future<bool> FileStream::closeAsync() {
    SYS_FAULT(eventGenerator_ != nullptr) << "该Stream没有附加到任何EventGenerator上,如果需要释放,请使用 delete";
    std::lock_guard<std::recursive_mutex> guard(dataLock_);
    //增加了 uv_is_closing 来确保不会被多次close ,但是仍然建议不要在多线程中不经过同步,就多次调用close
    Future<bool> closeFuture = closePromise_.getFuture();
    if (isClosing_ ||
        isClosed_ ||
        uv_is_closing((uv_handle_t *) &coreAsyncPollData.pollCtx)) {
        return closeFuture;
    }
    isClosing_ = YES;
    if (getCurrentThreadId() == coreAsyncPollData.tid) {
        //尤其是这里,如果已经关闭了,就不能执行如下命令
        FileStream::AsyncCloseCallBack(&closeAsyncCtx_);
    } else {
        //submit to task queue
        uv_async_send(&closeAsyncCtx_);
    }
    return closeFuture;
}

void FileStream::activate(EventFlag e) {
    if (isClosed_ ||
        isClosing_ ||
        //确保在关闭的时候,不会产生active的调用
        uv_is_closing((uv_handle_t *) &coreAsyncPollData.pollCtx)
            ) {
        return;
    }
    coreAsyncPollData.events = e;
    uv_poll_start(&coreAsyncPollData.pollCtx,
                  coreAsyncPollData.events,
                  &FileStream::DefaultIOCallBack);

}

void FileStream::attach(AsyncEventGenerator *gen) {
    AbstractAsyncStream::attach(gen);
    //这里将获取server所在tid,同时保存到socket的sockdata中
    //Server与Socket必然也是必须处于同一个线程中
    coreAsyncPollData.tid = gen->getEventGeneratorThreadId();
    //开始初始化异步执行,时间等等处理器
    //初始化Socket Loop
    CoreLoopContext *loopContext = eventGenerator_->getEventGeneratorCoreLoop();
    //初始化异步发送上下文
    uv_async_init(loopContext, &sendAsyncCtx_, &FileStream::AsyncSendCallBack);
    sendAsyncCtx_.data = this;
    ++openInHandle_;
    //初始化异步关闭上下文
    uv_async_init(loopContext, &closeAsyncCtx_, &FileStream::AsyncCloseCallBack);
    closeAsyncCtx_.data = this;
    ++openInHandle_;
    //初始化Idle上下文
    uv_timer_init(loopContext, &idleTimerCtx_);
    idleTimerCtx_.data = this;
    ++openInHandle_;
    //初始化pollContext
    SYS_FAULT(coreAsyncPollData.fd >= 0) << "文件描述符不合法!";
    uv_poll_init(loopContext, &coreAsyncPollData.pollCtx, coreAsyncPollData.fd);
    coreAsyncPollData.pollCtx.data = this;
    ++openInHandle_;
    //获得服务器对象,构建的套接字后,retain +1
    retain();
    //激活
    activate(IOEvent::READABLE);
    //标志打开
    OnOpen();
}

void FileStream::AsyncSendCallBack(AsyncContext *async) {
    FileStream *f = (FileStream *) async->data;
    f->activate(f->coreAsyncPollData.events | IOEvent::WRITABLE);
}

void FileStream::IdleTimerCallBack(TimerContext *timer) {
    //idle !
    FileStream *f = (FileStream *) timer->data;
#ifdef ENABLE_LOG
    LOG_I << "IDLE ! CLOSE SOCKET !";
#endif
    f->closeAsync();
}

void FileStream::AsyncCloseCallBack(uv_async_t *async) {
    FileStream *fs = (FileStream *) async->data;
    SYS_FAULT(&fs->closeAsyncCtx_ == async) << "非法调用 AsyncCloseCallBack";
    AsyncPollData &apd = fs->coreAsyncPollData;
    uv_timer_stop(&fs->idleTimerCtx_);
    uv_poll_stop(&apd.pollCtx);
    ::close(apd.fd);
    //close pollCtx
    uv_close((uv_handle_t *) &apd.pollCtx, &FileStream::UninstallHandlesCallBack);
    //close sendAsyncCtx
    uv_close((uv_handle_t *) &fs->sendAsyncCtx_, &FileStream::UninstallHandlesCallBack);
    //close idleTimerCtx
    uv_close((uv_handle_t *) &fs->idleTimerCtx_, &FileStream::UninstallHandlesCallBack);
    //close closeAsyncCtx
    uv_close((uv_handle_t *) &fs->closeAsyncCtx_, &FileStream::UninstallHandlesCallBack);
}

void FileStream::UninstallHandlesCallBack(uv_handle_t *h) {
    FileStream *fs = (FileStream *) h->data;
    SYS_FAULT(fs->isAttached()) << "fatal not attached";
    //当引用计数归0时,释放stream
    if (--fs->openInHandle_ == 0) {
        //关闭操作，和发送操作是互斥的
        {
            std::lock_guard<std::recursive_mutex> guard(fs->dataLock_);
            fs->OnClosed();
            fs->closePromise_.setValue(YES);
            std::list<IOBuffer> empty;
            fs->dataWillSend_.swap(empty);
        }
        fs->release();
    }
}

void FileStream::resetTimer() {
    if (idleTimeMs_ == 0) {
        //不超时,则永远不启动倒数计时器,但是将计时器挂载到loop上
        return;
    }
    uv_timer_start(&idleTimerCtx_, &FileStream::IdleTimerCallBack, idleTimeMs_, 0);
}

void FileStream::OnDataRead(const RawDataType *buf, size_t len) {
    resetTimer();
    ioBehavior_.notifyOnDataRead(this, buf, len);
}

void FileStream::OnOpen() {
    resetTimer();
    ioBehavior_.notifyOnOpen(this);
}

void FileStream::OnClosed() {
    isClosed_ = YES;
    ioBehavior_.notifyOnClosed(this);
}

void FileStream::DefaultIOCallBack(PollContext *handle, int status, EventFlag events) {
    FileStream *fstream = (FileStream *) handle->data;
    AbstractAsyncStream::AsyncPollData &pollData = fstream->coreAsyncPollData;
    if (status < 0) {
#ifdef ENABLE_LOG
        LOG_I << "clientIoCb status != 0 , client disconnected";
#endif
        fstream->closeAsync();
        return;
    }
    CHECK(events & pollData.events) << "clientIoCb events not contains callback events";
    CHECK(!(events & ~pollData.events)) << "clientIoCb events check failed";
    fstream->onStreamIO(status, events);
}
